package com.gmall

import java.net.InetSocketAddress
import java.util

import com.alibaba.fastjson.JSONObject
import com.alibaba.otter.canal.client.CanalConnectors
import com.alibaba.otter.canal.protocol.CanalEntry.{EntryType, EventType, RowChange}
import com.alibaba.otter.canal.protocol.{CanalEntry, Message}
import com.gmall.utils.{Constants, PropertiesUtil}
import com.google.protobuf.ByteString

import scala.collection.JavaConverters._
import scala.util.Random

object CanalClient {

    private val address: String = PropertiesUtil.getValue("canal.address")
    private val port: Int = PropertiesUtil.getValue("canal.port").toInt


    def main(args: Array[String]): Unit = {
        // 1、连接canal服务器
        val socketAddr: InetSocketAddress = new InetSocketAddress(address, port)
        // 1.1、创建canal服务器的连接器
        val connector = CanalConnectors.newSingleConnector(socketAddr, "example", "", "")
        // 1.2、连接到canal
        connector.connect()
        // 1.3、订阅数据，监控指定的表的数据的变化（.*所有表）
        connector.subscribe("gmall0523.*")
        // 2、从服务器获取数据
        while (true) {
            // 循环拉取数据 (一个消息对应多条sql语句的执行)
            // 100：一次最多获取100条sql导致的数据变化
            val msg: Message = connector.get(100)
            // 一个消息对应多行数据发生了变化, 一个entry表示一条sql语句的执行
            val entries: util.List[CanalEntry.Entry] = msg.getEntries

            if (entries.size() > 0) {
                // 遍历每行数据，并解析
                for (entry <- entries.asScala) {
                    // 过滤数据，只对ROWDATA这样的EntryType做处理
                    if (entry != null && entry.hasEntryType && entry.getEntryType == EntryType.ROWDATA) {
                        // 获取每行行数据，但是这种数据不是字符串，需要要解析
                        val storeValue: ByteString = entry.getStoreValue
                        val rowChange: RowChange = RowChange.parseFrom(storeValue)
                        val rowDataList: util.List[CanalEntry.RowData] = rowChange.getRowDatasList
                        handleRowData(entry.getHeader.getTableName, rowChange.getEventType, rowDataList)
                    }
                }
            } else {
                println("没有拉取到数据，3s后继续拉取。。。")
                Thread.sleep(3000)
            }
        }
        // 解析数据

        // 把解析的数据写入到kafka


    }

    /**
     * 解析rowData
     * @param tableName 表名
     * @param eventType 事件类型(插入, 删除等)
     * @param rowDataList 具体的数据
     */
    private def handleRowData(tableName: String, eventType: CanalEntry.EventType, rowDataList: util.List[CanalEntry.RowData]): Unit = {
        if (!rowDataList.isEmpty && tableName == "order_info" && eventType == EventType.INSERT) {
            handleData(rowDataList, Constants.ORDER_INFO_TOPIC)
        } else if (!rowDataList.isEmpty && tableName == "order_detail" && eventType == EventType.INSERT) {
            handleData(rowDataList, Constants.ORDER_DETAIL_TOPIC)
        }
    }

    private def handleData(rowDataList: util.List[CanalEntry.RowData], topic: String): Unit = {
        // rowData表示一行数据，遍历每一行数据
        for (rowData <- rowDataList.asScala) {
            // 获取每行中，所有的列
            val columns: util.List[CanalEntry.Column] = rowData.getAfterColumnsList
            val data: JSONObject = new JSONObject()
            // 将每个列的数据进行封装，封装成json对象
            for (column <- columns.asScala) {
                val key: String = column.getName
                val value: String = column.getValue
                data.put(key, value)
                println(key, value)
            }
            // 将数据发送到kafka
            new Thread() {
                override def run(): Unit = {
                    Thread.sleep(new Random().nextInt(10) * 1000)
                    MyKafkaUtil.sendToKafka(topic, data.toJSONString)
                }
            }.start()

        }
    }

}
